// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package org.apache.doris.datasource.property.storage;

import org.apache.doris.common.ExceptionChecker;
import org.apache.doris.common.UserException;

import com.google.common.collect.Maps;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;

import java.util.HashMap;
import java.util.Map;

public class OSSPropertiesTest {

    private static String ossAccessKey = "";
    private static String ossSecretKey = "";
    private static String hdfsPath = "";

    @Test
    public void testBasicCreateTest() {
        Map<String, String> origProps = new HashMap<>();
        origProps.put("oss.endpoint", "https://oss.aliyuncs.com");
        origProps.put("oss.access_key", "myOSSAccessKey");
        origProps.put("oss.secret_key", "myOSSSecretKey");
        origProps.put(StorageProperties.FS_OSS_SUPPORT, "true");
        Map<String, String> finalOrigProps = origProps;
        ExceptionChecker.expectThrowsWithMsg(IllegalArgumentException.class,
                "Region is not set. If you are using a standard endpoint, the region will be detected automatically. Otherwise, please specify it explicitly.", () -> StorageProperties.createPrimary(finalOrigProps));
        origProps.put("oss.endpoint", "oss-cn-shenzhen-finance-1-internal.aliyuncs.com");
        Map<String, String> finalOrigProps1 = origProps;
        OSSProperties ossProperties = (OSSProperties) StorageProperties.createPrimary(finalOrigProps1);
        Assertions.assertEquals("oss-cn-shenzhen-finance-1-internal.aliyuncs.com", ossProperties.getEndpoint());
        Assertions.assertEquals("cn-shenzhen-finance-1", ossProperties.getRegion());
        Assertions.assertDoesNotThrow(() -> StorageProperties.createPrimary(finalOrigProps1));
        origProps = new HashMap<>();
        origProps.put("oss.endpoint", "oss-cn-shenzhen-finance-1-internal.aliyuncs.com");
        Map<String, String> finalOrigProps2 = origProps;
        // allow both access_key and secret_key to be empty for anonymous access
        ExceptionChecker.expectThrowsNoException(() -> StorageProperties.createPrimary(finalOrigProps2));
    }

    @Test
    public void testToNativeS3Configuration() throws UserException {
        Map<String, String> origProps = new HashMap<>();
        origProps.put("oss.access_key", "myOSSAccessKey");
        origProps.put("oss.secret_key", "myOSSSecretKey");
        origProps.put("oss.endpoint", "oss-cn-beijing-internal.aliyuncs.com");
        origProps.put(StorageProperties.FS_OSS_SUPPORT, "true");
        origProps.put("oss.connection.maximum", "88");
        origProps.put("oss.connection.request.timeout", "100");
        origProps.put("oss.connection.timeout", "1000");
        origProps.put("oss.use_path_style", "true");
        origProps.put("test_non_storage_param", "6000");
        OSSProperties ossProperties = (OSSProperties) StorageProperties.createAll(origProps).get(0);
        Map<String, String> s3Props;

        Map<String, String> ossConfig = ossProperties.getMatchedProperties();
        Assertions.assertTrue(!ossConfig.containsKey("test_non_storage_param"));

        origProps.forEach((k, v) -> {
            if (!k.equals("test_non_storage_param") && !k.equals(StorageProperties.FS_OSS_SUPPORT)) {
                Assertions.assertEquals(v, ossConfig.get(k));
            }
        });


        s3Props = ossProperties.generateBackendS3Configuration();
        Assertions.assertEquals("oss-cn-beijing-internal.aliyuncs.com", s3Props.get("AWS_ENDPOINT"));
        Assertions.assertEquals("cn-beijing", s3Props.get("AWS_REGION"));
        Assertions.assertEquals("myOSSAccessKey", s3Props.get("AWS_ACCESS_KEY"));
        Assertions.assertEquals("myOSSSecretKey", s3Props.get("AWS_SECRET_KEY"));
        Assertions.assertEquals("88", s3Props.get("AWS_MAX_CONNECTIONS"));
        Assertions.assertEquals("100", s3Props.get("AWS_REQUEST_TIMEOUT_MS"));
        Assertions.assertEquals("1000", s3Props.get("AWS_CONNECTION_TIMEOUT_MS"));
        Assertions.assertEquals("true", s3Props.get("use_path_style"));
        origProps.remove("oss.use_path_style");
        ossProperties = (OSSProperties) StorageProperties.createAll(origProps).get(0);
        s3Props = ossProperties.generateBackendS3Configuration();
        Assertions.assertEquals("false", s3Props.get("use_path_style"));
    }

    @Test
    public void testGetRegion() throws UserException {
        Map<String, String> origProps = new HashMap<>();
        origProps.put("oss.endpoint", "oss-cn-hangzhou.aliyuncs.com");
        origProps.put("oss.access_key", "myOSSAccessKey");
        origProps.put("oss.secret_key", "myOSSSecretKey");
        OSSProperties ossProperties = (OSSProperties) StorageProperties.createPrimary(origProps);
        Assertions.assertEquals("cn-hangzhou", ossProperties.getRegion());
        Assertions.assertEquals("myOSSAccessKey", ossProperties.getAccessKey());
        Assertions.assertEquals("myOSSSecretKey", ossProperties.getSecretKey());
        Assertions.assertEquals("oss-cn-hangzhou.aliyuncs.com", ossProperties.getEndpoint());
        origProps.put("oss.endpoint", "oss-cn-hangzhou-internal.aliyuncs.com");
        Assertions.assertEquals("cn-hangzhou", ((OSSProperties) StorageProperties.createPrimary(origProps)).getRegion());
        origProps.put("oss.endpoint", "s3.oss-cn-shanghai.aliyuncs.com");
        Assertions.assertEquals("cn-shanghai", ((OSSProperties) StorageProperties.createPrimary(origProps)).getRegion());
        origProps.put("oss.endpoint", "s3.oss-cn-hongkong-internal.aliyuncs.com");
        Assertions.assertEquals("cn-hongkong", ((OSSProperties) StorageProperties.createPrimary(origProps)).getRegion());
        origProps.put("oss.endpoint", "https://s3.oss-cn-hongkong-internal.aliyuncs.com");
        Assertions.assertEquals("cn-hongkong", ((OSSProperties) StorageProperties.createPrimary(origProps)).getRegion());
        origProps.put("oss.endpoint", "http://s3.oss-cn-hongkong.aliyuncs.com");
        Assertions.assertEquals("cn-hongkong", ((OSSProperties) StorageProperties.createPrimary(origProps)).getRegion());
        origProps.put("oss.endpoint", "https://dlf.cn-beijing.aliyuncs.com");
        Assertions.assertEquals("cn-beijing", ((OSSProperties) StorageProperties.createAll(origProps).get(0)).getRegion());
        origProps.put("oss.endpoint", "datalake-vpc.cn-shenzhen.aliyuncs.com");
        Assertions.assertEquals("cn-shenzhen", ((OSSProperties) StorageProperties.createPrimary(origProps)).getRegion());
        origProps.put("oss.endpoint", "https://datalake-vpc.cn-shenzhen.aliyuncs.com");
        Assertions.assertEquals("cn-shenzhen", ((OSSProperties) StorageProperties.createPrimary(origProps)).getRegion());
        origProps.put("oss.endpoint", "http://datalake-vpc.eu-central-1.aliyuncs.com");
        Assertions.assertEquals("eu-central-1", ((OSSProperties) StorageProperties.createPrimary(origProps)).getRegion());
    }

    @Test
    public void testGetRegionWithDefault() throws UserException {
        Map<String, String> origProps = new HashMap<>();
        origProps.put("uri", "https://examplebucket-1250000000.oss-cn-hangzhou.aliyuncs.com/test/file.txt");
        origProps.put("oss.access_key", "myOSSAccessKey");
        origProps.put("oss.secret_key", "myOSSSecretKey");
        OSSProperties ossProperties = (OSSProperties) StorageProperties.createPrimary(origProps);
        Assertions.assertEquals("cn-hangzhou", ossProperties.getRegion());
        Assertions.assertEquals("myOSSAccessKey", ossProperties.getAccessKey());
        Assertions.assertEquals("myOSSSecretKey", ossProperties.getSecretKey());
        Assertions.assertEquals("oss-cn-hangzhou.aliyuncs.com", ossProperties.getEndpoint());
        Map<String, String> ossNoEndpointProps = new HashMap<>();
        ossNoEndpointProps.put("oss.access_key", "myOSSAccessKey");
        ossNoEndpointProps.put("oss.secret_key", "myOSSSecretKey");
        ossNoEndpointProps.put("oss.region", "cn-hangzhou");
        origProps.put("uri", "s3://examplebucket-1250000000/test/file.txt");
        // oss support without endpoint
        ExceptionChecker.expectThrowsWithMsg(IllegalArgumentException.class, "Endpoint is not set. Please specify it explicitly.", () -> StorageProperties.createPrimary(ossNoEndpointProps));
    }

    @Test
    public void testMissingAccessKey() {
        Map<String, String> origProps = new HashMap<>();
        origProps.put("oss.endpoint", "oss-cn-hangzhou.aliyuncs.com");
        origProps.put("oss.secret_key", "myOSSSecretKey");
        ExceptionChecker.expectThrowsWithMsg(IllegalArgumentException.class,
                "Both the access key and the secret key must be set.",
                () -> StorageProperties.createPrimary(origProps));
        origProps.remove("oss.secret_key");
        Assertions.assertDoesNotThrow(() -> StorageProperties.createPrimary(origProps));
    }

    @Test
    public void testDlfProperties() {
        Map<String, String> origProps = new HashMap<>();
        origProps.put("iceberg.catalog.type", "dlf");
        origProps.put("dlf.region", "cn-beijing");
        origProps.put("dlf.access.public", "true");
        OSSProperties ossProperties = OSSProperties.of(origProps);
        Assertions.assertEquals("oss-cn-beijing.aliyuncs.com", ossProperties.getEndpoint());
    }

    @Test
    public void testMissingSecretKey() {
        Map<String, String> origProps = new HashMap<>();
        origProps.put("oss.endpoint", "oss-cn-hangzhou.aliyuncs.com");
        origProps.put("oss.access_key", "myOSSAccessKey");
        ExceptionChecker.expectThrowsWithMsg(IllegalArgumentException.class,
                "Both the access key and the secret key must be set.",
                () -> StorageProperties.createPrimary(origProps));
        origProps.remove("oss.access_key");
        Assertions.assertDoesNotThrow(() -> StorageProperties.createPrimary(origProps));
    }

    @Test
    public void testDlfPropertiesEndpoint() {
        Map<String, String> origProps = new HashMap<>();
        origProps.put("type", "iceberg");
        origProps.put("warehouse", "oss://bucket/hive-dlf-oss-warehouse/iceberg/dlf-oss/");
        origProps.put("dlf.region", "cn-beijing");
        origProps.put("dlf.endpoint", "datalake-vpc.cn-beijing.aliyuncs.com");
        origProps.put("dlf.uid", "12345");
        origProps.put("dlf.catalog.id", "p2_regression_case");
        origProps.put("dlf.access_key", "ACCESS_KEY");
        origProps.put("dlf.secret_key", "SECERT_KET");
        origProps.put("dlf.access.public", "true");
        OSSProperties ossProperties = OSSProperties.of(origProps);
        Assertions.assertEquals("oss-cn-beijing.aliyuncs.com", ossProperties.getEndpoint());
        origProps.remove("dlf.access.public");
        ossProperties = OSSProperties.of(origProps);
        Assertions.assertEquals("oss-cn-beijing-internal.aliyuncs.com", ossProperties.getEndpoint());
        origProps.put("oss.endpoint", "dlf.cn-beijing.aliyuncs.com");
        ossProperties = OSSProperties.of(origProps);
        Assertions.assertEquals("oss-cn-beijing-internal.aliyuncs.com", ossProperties.getEndpoint());
        origProps.put("oss.endpoint", "dlf-vpc.cn-beijing.aliyuncs.com");
        ossProperties = OSSProperties.of(origProps);
        Assertions.assertEquals("oss-cn-beijing-internal.aliyuncs.com", ossProperties.getEndpoint());
    }


    @Test
    public void testNotEndpoint() throws UserException {
        Map<String, String> origProps = new HashMap<>();
        origProps.put("uri", "oss://examplebucket-1250000000/test/file.txt");
        origProps.put("oss.access_key", "myOSSAccessKey");
        origProps.put("oss.secret_key", "myOSSSecretKey");
        origProps.put("oss.region", "cn-hangzhou");
        Assertions.assertEquals("oss-cn-hangzhou-internal.aliyuncs.com",
                ((OSSProperties) StorageProperties.createPrimary(origProps)).getEndpoint());
        origProps.put("dlf.access.public", "true");
        Assertions.assertEquals("oss-cn-hangzhou.aliyuncs.com",
                ((OSSProperties) StorageProperties.createPrimary(origProps)).getEndpoint());
        origProps.put("uri", "https://doris-regression-hk.oss-cn-hangzhou-internal.aliyuncs.com/regression/datalake/pipeline_data/data_page_v2_gzip.parquet");
        Assertions.assertEquals("oss-cn-hangzhou-internal.aliyuncs.com", ((OSSProperties) StorageProperties.createPrimary(origProps)).getEndpoint());
    }

    @Test
    public void testOSSProperties() throws UserException {
        Map<String, String> origProps = new HashMap<>();
        origProps.put("warehouse", "new_dlf_paimon_catalog");
        origProps.put("uri", "http://cn-beijing-vpc.dlf.aliyuncs.com");
        origProps.put("type", "paimon");
        origProps.put("paimon.rest.token.provider", "dlf");
        origProps.put("paimon.rest.dlf.access-key-secret", "XXXXX");
        origProps.put("paimon.rest.dlf.access-key-id", "XXXXXX");
        origProps.put("paimon.catalog.type", "rest");
        Assertions.assertEquals(1, StorageProperties.createAll(origProps).size());
    }

    @Test
    public void testAwsCredentialsProvider() throws Exception {
        Map<String, String> ossProps = new HashMap<>();
        ossProps.put("fs.oss.support", "true");
        ossProps.put("oss.endpoint", "oss-cn-hangzhou.aliyuncs.com");
        OSSProperties ossStorageProperties = (OSSProperties) StorageProperties.createPrimary(ossProps);
        Assertions.assertEquals(AnonymousCredentialsProvider.class, ossStorageProperties.getAwsCredentialsProvider().getClass());
        ossProps.put("oss.access_key", "myAccessKey");
        ossProps.put("oss.secret_key", "mySecretKey");
        ossStorageProperties = (OSSProperties) StorageProperties.createPrimary(ossProps);
        Assertions.assertEquals(StaticCredentialsProvider.class, ossStorageProperties.getAwsCredentialsProvider().getClass());
    }

    @Test
    public void testS3DisableHadoopCache() throws UserException {
        Map<String, String> props = Maps.newHashMap();
        props.put("oss.endpoint", "oss-cn-hangzhou.aliyuncs.com");
        OSSProperties s3Properties = (OSSProperties) StorageProperties.createPrimary(props);
        Assertions.assertTrue(s3Properties.hadoopStorageConfig.getBoolean("fs.oss.impl.disable.cache", false));
        props.put("fs.oss.impl.disable.cache", "true");
        s3Properties = (OSSProperties) StorageProperties.createPrimary(props);
        Assertions.assertTrue(s3Properties.hadoopStorageConfig.getBoolean("fs.oss.impl.disable.cache", false));
        props.put("fs.oss.impl.disable.cache", "false");
        s3Properties = (OSSProperties) StorageProperties.createPrimary(props);
        Assertions.assertFalse(s3Properties.hadoopStorageConfig.getBoolean("fs.oss.impl.disable.cache", false));
        props.put("fs.oss.impl.disable.cache", "null");
        s3Properties = (OSSProperties) StorageProperties.createPrimary(props);
        Assertions.assertFalse(s3Properties.hadoopStorageConfig.getBoolean("fs.oss.impl.disable.cache", false));
    }

}
